package storm;


import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.util.Iterator;
import java.util.List;

import utilities.Suggester;
import utilities.Twokenizer;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class SuggesterBolt extends BaseBasicBolt {

	private static final long serialVersionUID = 4493364795064805398L;
	
	@Override
	public void execute(Tuple input, BasicOutputCollector collector) {
		String tweet = input.getString(0);
		Twokenizer tokenizer = new Twokenizer();
		List<String> tokens = tokenizer.twokenize(tweet);
		Suggester suggester = new Suggester();
		String tmp;
		String sugg;
		String tweetSugg = "";
		Iterator<String> iter = tokens.iterator();
		while(iter.hasNext()){
			tmp = ((String)iter.next());
			if(tmp.startsWith("#")||tmp.startsWith("@")||tmp.contains("http")){
				tweetSugg = tweetSugg + " " + tmp;
			}
			else{
				sugg = suggester.wordSuggester(tmp);
				if(sugg.equals("")){
					 tweetSugg = tweetSugg + " " + tmp;
				 }
				 else{
					 tweetSugg = tweetSugg + " " + sugg;
				 }
			}
		}
		collector.emit(new Values(tweetSugg));
	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("tweetSugg"));
	}

	public void cleanup() {
		createFile();
		System.out.println("CREATED DUMP FILE");
		
	}
	private void createFile(){
		PrintStream ps=null;
		try {
			ps = new PrintStream(new FileOutputStream("output.txt"));
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		ps.close();
	}
}
